package connect.service;

import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.ChannelDuplexHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;
import io.netty.util.internal.ConcurrentSet;
import java.util.Iterator;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import connect.Decoder;
import connect.Encoder;
import connect.Message;

public class CommunicationManager{
	
	private ConcurrentSet<Connection> list=new ConcurrentSet<Connection>();//在线连接列表
	
	/**
	 * 构造
	 * @param service
	 * @param listener
	 */
	public CommunicationManager(final Service service,final ResponseListener listener) {
		
		ExecutorService pool=Executors.newCachedThreadPool();
		
		final Repeater repeater=new Repeater(list);
		
		final Client message=new Client(list);
		
		
		pool.execute(new Runnable() {
			
			public void run() {
				
				EventLoopGroup bossGroup=new NioEventLoopGroup();
				EventLoopGroup workerGroup=new NioEventLoopGroup();
				
				try {
					
					ServerBootstrap b=new ServerBootstrap();
					b.group(bossGroup, workerGroup)
					.channel(NioServerSocketChannel.class)
					.option(ChannelOption.SO_BACKLOG, 1024)
					.childHandler(new ChannelInitializer<SocketChannel>() {
						@Override
						protected void initChannel(SocketChannel arg0) throws Exception {
							
							arg0.pipeline().addLast(new Encoder());
							arg0.pipeline().addLast(new Decoder());
							
							arg0.pipeline().addLast("idleStateHandler", new IdleStateHandler(10,5,0));//心跳
							
							arg0.pipeline().addLast("HeartbeatHandler", new HeartbeatHandler());
							
							
							
							arg0.pipeline().addLast(new ChannelInboundHandlerAdapter(){
								
								
								
								@Override
								public void channelActive(ChannelHandlerContext ctx) 
										throws Exception {
									
									/**
									 * 连接规则
									 * 
									 * 同一个ip地址
									 * 不允许超过50条连接
									 * 否则 拒绝连接
									 * 
									 * 防御高并发连接攻击
									 */
									
									int size=0;
									
									String[] ip=String.valueOf(ctx.channel().remoteAddress()).split(":");
									
									Iterator<Connection> it=list.iterator();
									
									while (it.hasNext()) {
										Connection c=it.next();
										if (ip[0].equals(c.getIp())) {
											size++;
										}
									}
									
									if (size<50) {
										list.add(new Connection(ctx));
									}else {
										ctx.disconnect();
									}
								};
								
								
								
								@Override
								public void channelRead(ChannelHandlerContext ctx,
										Object msg) throws Exception {
									 
									if (listener!=null) {
										message.setMessage(msg);
										message.setContext(ctx);
										listener.OnMessage(repeater,message);
									}	
									
								}
								
								public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) 
										throws Exception {
										
									
								};
								
								@Override
								public void handlerRemoved(
										ChannelHandlerContext ctx)
										throws Exception {
										
										Iterator<Connection> it=list.iterator();
										
										while (it.hasNext()) {
											Connection c=it.next();
											if (c.getContext()==ctx) {
												if (listener!=null&&c.getKey().length()>0) {
													listener.OnBreak(c.getKey());
												}
												it.remove();
											}
											
										}
										ctx.close();
									}
								});
							}
						});
					
					try {
						
						b.bind(service.getPort()).channel().closeFuture().sync();;
						
					} catch (InterruptedException e) {
						
					}
					
				} finally {
					//退出 释放线程池资源
					bossGroup.shutdownGracefully();
					workerGroup.shutdownGracefully();
					
				}
				
			}
		});
		pool.shutdown();
		
	}
	
	class HeartbeatHandler extends ChannelDuplexHandler {
		
		@Override
		public void channelRead(ChannelHandlerContext ctx, Object msg)
				throws Exception {
			
			if (msg instanceof Message) {
				ctx.fireChannelRead(msg);
			}else if (msg instanceof ByteBuf) {
				ByteBuf buf=Unpooled.buffer();
            	buf.writeByte(0);
            	buf.writeInt(0);
            	ctx.writeAndFlush(buf);
			}
			
		}
		
		@Override  
	    public void userEventTriggered(ChannelHandlerContext ctx, Object evt)  
	            throws Exception {  
	        if (evt instanceof IdleStateEvent) {  
	            IdleStateEvent e = (IdleStateEvent) evt;  
	            if (e.state() == IdleState.ALL_IDLE) {  
	                ctx.close();  
	            } 
	        }  
	    }  
	 }
}
